package com.bw.ads;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.fun.Mn6AsyncIOFunc2;
import com.bw.fun.Mn6AsyncIOFunc3;
import com.bw.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple9;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class Test10 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk6","test695"));

        // Order_Info
        SingleOutputStreamOperator<String> oiDS = stream
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(s);
                            String table = jsonObject.getString("table");
                            if ("order_info".equals(table) && jsonObject.getJSONObject("data").getString("status") != null && jsonObject.getJSONObject("data").getString("status").equals("60050")) {
                                return true;
                            }
                        } catch (Exception e) {
                        }
                        return false;
                    }
                });

        // transport_task
        SingleOutputStreamOperator<String> ttDS = stream
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(s);
                            String table = jsonObject.getString("table");
                            if ("transport_task".equals(table) && jsonObject.getJSONObject("data").getString("status") != null && jsonObject.getJSONObject("data").getString("status").equals("67004")) {
                                return true;
                            }
                        } catch (Exception e) {
                        }
                        return false;
                    }
                });

        // 把Order_Info轉成9元組
        // 日期,城市,下單金額，下單次數,運輸次數,運輸里程，运输时长，每次平均运输时长，平均每次运输？
        SingleOutputStreamOperator<Tuple9<String, String, Double, Integer, Integer, Double, Long, Double, Double>> u1 =
                AsyncDataStream.unorderedWait(oiDS, new Mn6AsyncIOFunc2(), 1000, TimeUnit.MILLISECONDS, 100);

        //日期,城市,下單金額，下單次數,運輸次數,運輸里程，运输时长，每次平均运输时长，平均每次运输？
        SingleOutputStreamOperator<Tuple9<String, String, Double, Integer, Integer, Double, Long, Double, Double>> u2 = AsyncDataStream.unorderedWait(ttDS, new Mn6AsyncIOFunc3(), 1000, TimeUnit.MILLISECONDS, 100);

        // 链接流
        DataStream<Tuple9<String, String, Double, Integer, Integer, Double, Long, Double, Double>> union = u1.union(u2);

        SingleOutputStreamOperator<Tuple9<String, String, Double, Integer, Integer, Double, Long, Double, Double>> reduce = union.keyBy(new KeySelector<Tuple9<String, String, Double, Integer, Integer, Double, Long, Double, Double>, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(Tuple9<String, String, Double, Integer, Integer, Double, Long, Double, Double> v) throws Exception {
                return new Tuple2<>(v.f0, v.f1);
            }
        }).reduce(new ReduceFunction<Tuple9<String, String, Double, Integer, Integer, Double, Long, Double, Double>>() {
            @Override
            public Tuple9<String, String, Double, Integer, Integer, Double, Long, Double, Double> reduce(Tuple9<String, String, Double, Integer, Integer, Double, Long, Double, Double> x, Tuple9<String, String, Double, Integer, Integer, Double, Long, Double, Double> y) throws Exception {
                double d1 = 0.0;
                double d2 = 0.0;
                if ((x.f5 + y.f5) != 0.0 && (x.f4 + y.f4) != 0) {
                    d1 = (x.f5 + y.f5) / (x.f4 + y.f4);
                }
                if ((x.f6 + y.f6) != 0L && (x.f4 + y.f4) != 0) {
                    d2 = Double.valueOf(x.f6 + y.f6) / (x.f4 + y.f4);
                }
                return new Tuple9<>(x.f0, x.f1, x.f2 + y.f2, x.f3 + y.f3, x.f4 + y.f4, x.f5 + y.f5, x.f6 + y.f6, d1, d2);
            }
        });

        reduce.print();
        env.execute();
    }
}
